package ldh.im.fxbase.thread;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class DynamicThreadPool implements Executor {

    private Logger LOGGER = LoggerFactory.getLogger(DynamicThreadPool.class);

    private int coreThreadSize = 3;
    private long activeTime;
    private TimeUnit activeTimeUtil;
    private Thread[] threads = null;
    private BlockingQueue<Runnable> blockingQueue;
    private Lock executeLock = new ReentrantLock();
    private static AtomicInteger THREAD_INDEX = new AtomicInteger(1);

    public DynamicThreadPool(int coreThreadSize, long activeTime, TimeUnit activeTimeUtil) {
        if (coreThreadSize < 1) {
            throw new IllegalArgumentException("coreThreadSize不能小于1");
        }
        if (coreThreadSize > 100) {
            throw new IllegalArgumentException("coreThreadSize不能大于100");
        }
        this.coreThreadSize = coreThreadSize;
        this.activeTime = activeTime;
        this.activeTimeUtil = activeTimeUtil;
        threads = new Thread[coreThreadSize];
        blockingQueue = new ArrayBlockingQueue<>(10000);
    }

    @Override
    public void execute(Runnable runnable) {
        blockingQueue.add(runnable);
        executeLock.lock();
        try {
            startIdleStart();
        } finally {
            executeLock.unlock();
        }

    }

    public void close() {
        executeLock.lock();
        try {
            for (int i=0; i<coreThreadSize; i++) {
                if (threads[i] != null && threads[i].isAlive()) {
                    threads[i].interrupt();
                }
            }
        } finally {
            executeLock.unlock();
        }
    }

    private void startIdleStart() {
        for (int i=0; i<coreThreadSize; i++) {
            if (threads[i] == null) {
                threads[i] = new BusinessThread(blockingQueue, activeTime, activeTimeUtil);
                threads[i].start();
                break;
            } else if (threads[i].getState() == Thread.State.TERMINATED) {
                threads[i] = new BusinessThread(blockingQueue, activeTime, activeTimeUtil);
                threads[i].start();
                LOGGER.info("thread[{}] restart", i);
                break;
            }
        }
    }

    class BusinessThread extends Thread {

        private BlockingQueue<Runnable> blockingQueue;
        private long time;
        private TimeUnit timeUnit;

        public BusinessThread(BlockingQueue blockingQueue, long time, TimeUnit timeUnit) {
            super("DynamicThreadPool-" + THREAD_INDEX.incrementAndGet());
            this.blockingQueue = blockingQueue;
            this.time = time;
            this.timeUnit = timeUnit;
        }

        @Override
        public void run() {
            while(true) {
                try {
                    Runnable runnable = blockingQueue.poll(time, timeUnit);
                    if (runnable == null) {
                        break;
                    }
                    try {
                        runnable.run();
                    } catch (Exception e) {
                        LOGGER.error("处理错误", e);
                    }
                } catch (InterruptedException i) {
                    Thread.currentThread().interrupt();
                }
            }
            LOGGER.info("线程销毁");
        }
    }
}